package com.whoami.idmagic.apprank

import org.apache.spark.sql.{Encoder, Encoders, Row, SparkSession}

import scala.collection.{immutable, mutable}
import scala.collection.mutable.ListBuffer

object Sample {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().enableHiveSupport().getOrCreate()
    implicit val enc: Encoder[Row] = Encoders.kryo[Row]
    import org.apache.spark.sql.functions._
    import spark.implicits._

    val apps: List[String] = List("微信", "支付宝", "淘宝", "QQ", "亿通行", "高德地图", "京东", "美团", "WPS Office",
      "腾讯视频", "爱奇艺", "新浪微博", "抖音", "滴滴出行", "百度", "百度地图", "百度网盘", "招商银行", "拼多多", "优酷",
      "QQ浏览器", "铁路12306", "华为应用市场", "网易云音乐", "大众点评", "携程旅行", "QQ音乐", "墨迹天气", "趣头条", "pptv",
      "风行视频", "掌阅", "斗鱼直播", "妈妈网孕育", "火山小视频", "西瓜视频")
    val targetPercentage: List[Double] = List(0.915, 0.529, 0.516, 0.540, 0.002, 0.313, 0.210, 0.101, 0.072, 0.445,
      0.507, 0.409, 0.304, 0.060, 0.358, 0.246, 0.043, 0.026, 0.282, 0.370, 0.317, 0.052, 0.135, 0.113, 0.058, 0.052,
      0.246, 0.130, 0.066, 0.023, 0.025, 0.088, 0.027, 0.008, 0.131, 0.133)

    val odsAplWRaw = spark.read.orc("/user/mz_spt/datalake/src/applist/ods_apl_w/20190812_20190818")
      .select($"md5_imei", $"applist")
      .filter(_.getString(0) != "")
      .dropDuplicates("md5_imei")
      .repartition(30)

    val aplBabelMap: mutable.Map[String, String] = mutable.Map()
    val aplBabel = spark.read.orc("/user/mz_spt/datalake/src/babel/dim_apl_babel/orc/20190821").select($"app_name", $"app_alias")
    val aplBabelIterator = aplBabel.toLocalIterator()
    while (aplBabelIterator.hasNext) {
      val row = aplBabelIterator.next()
      val appAlias = row.getString(1).split(",")
      val appName = row.getString(0)
      for (e <- appAlias) {
        aplBabelMap.put(e, appName)
      }
    }
    val odsAplWMid = odsAplWRaw.map(row => {
      val md5Imei = row.getString(0)
      val appList = row.getAs[mutable.WrappedArray[immutable.Map[String, String]]](1)
      val result: ListBuffer[Int] = ListBuffer()
      for (_ <- 1 to 36) {
        result.append(0)
      }
      for (e <- appList) {
        val packageName = e("package")
        val appName = aplBabelMap.get(packageName)
        if (appName.isDefined) {
          val index = apps.indexOf(appName.get)
          if (index >= 0) {
            result(index) = 1
          }
        }
      }
      (md5Imei, result)
    })
    var rawPercentageCount = odsAplWMid.reduce((a, b) => {
      val result: mutable.ListBuffer[Int] = mutable.ListBuffer()
      for (i <- a._2.indices) {
        result.append(a._2(i) + b._2(i))
      }
      (a._1, result)
    })._2
    var count = odsAplWMid.count
    var rawPercentage = rawPercentageCount.map(x => x.toFloat / count)
    println(rawPercentage)

    var diffPercentage = rawPercentage.map(x => x - targetPercentage(rawPercentage.indexOf(x)))
    diffPercentage = diffPercentage.map(x => (x - diffPercentage.min) / (diffPercentage.max - diffPercentage.min))
    var odsAplWValue = odsAplWMid.map(x => {
      var value = 0d
      for (i <- x._2.indices) {
        value += math.pow(diffPercentage(i) - x._2(i), 2)
      }
      (x._1, x._2, value)
    }).orderBy(desc("_3")).limit((count * 0.9).toInt)
    rawPercentageCount = odsAplWValue.reduce((a, b) => {
      val result: mutable.ListBuffer[Int] = mutable.ListBuffer()
      for (i <- a._2.indices) {
        result.append(a._2(i) + b._2(i))
      }
      (a._1, result, a._3)
    })._2
    count = (count * 0.9).toInt
    rawPercentage = rawPercentageCount.map(x => x.toFloat / count)
    println(rawPercentage)

    for (_ <- 0 to 10) {
      diffPercentage = rawPercentage.map(x => x - targetPercentage(rawPercentage.indexOf(x)))
      diffPercentage = diffPercentage.map(x => (x - diffPercentage.min) / (diffPercentage.max - diffPercentage.min))
      odsAplWValue = odsAplWValue.map(x => {
        var value = 0d
        for (i <- x._2.indices) {
          value += math.pow(diffPercentage(i) - x._2(i), 2)
        }
        (x._1, x._2, value)
      }).orderBy(desc("_3")).limit((count * 0.9).toInt)
      rawPercentageCount = odsAplWValue.reduce((a, b) => {
        val result: mutable.ListBuffer[Int] = mutable.ListBuffer()
        for (i <- a._2.indices) {
          result.append(a._2(i) + b._2(i))
        }
        (a._1, result, a._3)
      })._2
      count = (count * 0.9).toInt
      rawPercentage = rawPercentageCount.map(x => x.toFloat / count)
      println(rawPercentage)
    }
  }
}
